package bsp.scan;

import java.io.IOException;
import java.util.StringTokenizer;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hama.bsp.BSPPeer;
import org.apache.hama.bsp.sync.SyncException;

import util.ScanMessage;
import util.functors.OpScan;
import bsp.CMapper;

public class ScanStepOne<T extends Number> extends
		CMapper<LongWritable, Text, Text, NullWritable, ScanMessage> {

	private long chunkey = -1;
	private int myId;
	private String myName;
	private int numPeers;
	private OpScan<T> plus; //OpScan<T> : vb1
	
	public ScanStepOne(OpScan<T> op){
		plus = op;
	}
	@Override
	public void map(
			BSPPeer<LongWritable, Text, Text, NullWritable, ScanMessage> peer)
			throws IOException, SyncException, InterruptedException {

		LongWritable key = new LongWritable(-1);
		// only support long
		T sum =  plus.id();
		Text line = new Text();
		// StringBuffer newList = new StringBuffer(512);
		while (peer.readNext(key, line)) {
			if (chunkey < 0)
				chunkey = key.get();
			StringTokenizer itr = new StringTokenizer(line.toString().trim());
			while (itr.hasMoreTokens()) {
				sum = plus.evaluate(sum, str2Num(itr.nextToken()));
			}
		}
		// send p-sum to all
		// key: last line-key

		if (chunkey >= 0) {
			int peerId = myId + 1;
			while (peerId < numPeers) {
				addMesg(peer.getPeerName(peerId), new ScanMessage(chunkey, sum.toString()));
				System.out.println(myName + " sends to "
						+ peer.getPeerName(peerId));
				peerId++;
			}
		}
	}
	
	protected String Num2Str(T val){
		return val.toString();
	}
	protected double str2Num(String token){
		return  Double.parseDouble(token);
	}
}